package pers.mihao.quickstream.common.task.manager.impl;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import org.springframework.stereotype.Component;
import pers.mihao.quickstream.common.task.constans.TaskStatus;
import pers.mihao.quickstream.common.task.constans.TaskType;
import pers.mihao.quickstream.common.task.domain.runner.JavaAutomaticTaskRunner;
import pers.mihao.quickstream.common.task.domain.scheduler.AutomaticTaskTimer;
import pers.mihao.quickstream.common.task.dto.AutomaticTaskDTO;
import pers.mihao.quickstream.common.task.dto.JavaAutomaticTaskDTO;
import pers.mihao.quickstream.common.task.dto.UniqueBizInfo;
import pers.mihao.quickstream.common.task.manager.AutomaticTaskManager;
import pers.mihao.quickstream.common.task.repository.AutomaticTaskRepository;
import pers.mihao.quickstream.common.task.scheduler.SchedulerCenter;
import pers.mihao.quickstream.common.task.scheduler.TaskInvokeRes;
import pers.mihao.quickstream.common.util.CollectionUtil;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static pers.mihao.quickstream.common.task.constans.FeatureKey.*;
import static pers.mihao.quickstream.common.task.constans.TaskStatus.STOP;
import static pers.mihao.quickstream.common.task.constans.TaskStatus.TOBE_SCHEDULER;

/**
 * @author mihao
 * @since 2024/11/17
 */
@Component
public class AutomaticTaskManagerImpl implements AutomaticTaskManager {

    @Resource
    AutomaticTaskRepository automaticTaskRepository;

    @Resource
    SchedulerCenter schedulerCenter;

    ScheduledThreadPoolExecutor callbackExecutor = new ScheduledThreadPoolExecutor(1, r -> {
        Thread thread = new Thread(r);
        thread.setName("callbackExecutor");
        return thread;
    });


    @Override
    public AutomaticTaskDTO saveAutomaticTask(JavaAutomaticTaskDTO<?> javaTaskDTO) {
        AutomaticTaskTimer automaticTaskTimer = javaTaskDTO.getAutomaticTaskTimer();
        UniqueBizInfo uniqueBizInfo = javaTaskDTO.getUniqueBizInfo();
        JavaAutomaticTaskRunner javaTaskRunner = new JavaAutomaticTaskRunner(javaTaskDTO.getDoTaskInvoker(),
                javaTaskDTO.getJobParams());
        AutomaticTaskDTO task = AutomaticTaskDTO.builder()
                .bizId(uniqueBizInfo.getBizId())
                .bizType(uniqueBizInfo.getBizType())
                .name(javaTaskDTO.getDoTaskInvoker().getName())
                .description(javaTaskDTO.getDoTaskInvoker().getName())
                .status(TOBE_SCHEDULER)
                .priority(javaTaskDTO.getPriority())
                .retryCount(javaTaskDTO.getRetryCount())
                .timeLimit(javaTaskDTO.getTimeLimit())
                .nextInvokeTime(automaticTaskTimer.getNextInvokeTime(null, 0))
                .schedulerTimeType(automaticTaskTimer.type())
                .schedulerTimeValue(JSONObject.toJSONString(automaticTaskTimer, SerializerFeature.WriteClassName))
                .taskType(TaskType.JAVA)
                .taskConfig(JSONObject.toJSONString(javaTaskRunner, SerializerFeature.WriteClassName))
                .version(1)
                .build();
        automaticTaskRepository.saveAutomaticTask(task);
        onCreateCallback(task);
        return task;
    }

    private void onCreateCallback(AutomaticTaskDTO task) {
        if (callbackExecutor.getQueue().isEmpty()) {
            // 100ms最多回告一次
            callbackExecutor.schedule(() -> schedulerCenter.onNewTaskCreate(),
                    100, TimeUnit.MILLISECONDS);
        }
    }


    /**
     * 获取待调度的任务，改成调度中
     *
     * @param timeRange
     * @return
     */
    @Override
    public List<AutomaticTaskDTO> pullTobeSchedulerTask(Long timeRange) {
        List<AutomaticTaskDTO> taskDTOS = automaticTaskRepository.queryWaitToScheduler(timeRange);
        if (CollectionUtil.isEmpty(taskDTOS)) {
            return List.of();
        }
        for (AutomaticTaskDTO taskDTO : taskDTOS) {
            taskDTO.setStatus(TaskStatus.SCHEDULER_ING);
        }
        automaticTaskRepository.batchUpdateStatus(taskDTOS);
        return taskDTOS;
    }

    @Override
    public void taskInvoke(AutomaticTaskDTO taskDTO, TaskInvokeRes taskInvokeRes) {
        // 记录上次执行结果
        Optional.ofNullable(taskInvokeRes.getError()).map(Throwable::getMessage).ifPresent(taskInvokeRes::setErrorMessage);
        taskDTO.setLastInvokeTime(taskInvokeRes.getTaskStartTime());
        Integer num = taskDTO.getFeatures().readIntOrDefault(invokeNum, 0);
        taskDTO.getFeatures().update(lastInvokeRes, JSONObject.toJSONString(taskInvokeRes));
        taskDTO.getFeatures().update(invokeNum, String.valueOf(num + 1));
        AutomaticTaskTimer automaticTaskTimer = JSONObject.parseObject(taskDTO.getSchedulerTimeValue(), AutomaticTaskTimer.class);
        Date nextInvokeTime = automaticTaskTimer.getNextInvokeTime(taskInvokeRes, num);
        if (nextInvokeTime != null && nextInvokeTime.getTime() >= System.currentTimeMillis()) {
            taskDTO.setNextInvokeTime(nextInvokeTime);
            taskDTO.setStatus(TOBE_SCHEDULER);
        } else {
            taskDTO.getFeatures().update(stopReason, "没有执行时间停止");
            taskDTO.setStatus(STOP);
        }
        automaticTaskRepository.saveAutomaticTask(taskDTO);
    }
}
